package com.landcode.land.data.canal.manager;

import com.alibaba.otter.canal.protocol.Message;
import com.landcode.land.data.SynDataHandler;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author Alan
 */
@Component
@Slf4j
public class CanalCoreManager extends AbstractCanalManager {

    @Resource
    protected SynDataHandler synDataHandler;

    @Override
    public void process() {
        while (true) {
            try {
                MDC.put("destination", config.getDestination());
                canalConnector.connect();
                canalConnector.subscribe();
                while (true) {
                    // 获取指定数量的数据
                    Message message = canalConnector.getWithoutAck(config.getBatchSize());
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    // 没有要同步的数据
                    if (batchId == -1 || size == 0) {
                        try {
                            Thread.sleep(config.getWaitingTime());
                        } catch (InterruptedException e) {

                        }
                        continue;
                    } else {
                        // 打印数据
                        printMsg(message, batchId, size);
                        // 同步数据
                        synData(message);
                    }
                    // 提交确认
                    canalConnector.ack(batchId);
                    // connector.rollback(batchId); // 处理失败, 回滚数据
                }
            } catch (Exception e) {
                log.error("canal process error!", e);
            } finally {
                canalConnector.disconnect();
                MDC.remove("destination");
            }
        }
    }

    @Override
    public void synData(Message message) throws Exception {
        synDataHandler.synData(message);
    }
}
